package com.ruoyi.eqmonitor.dispose.task;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ObjectUtil;
import com.ruoyi.common.redis.service.RedisService;
import com.ruoyi.eqmonitor.dispose.mapper.MonitorDataMapper;
import com.ruoyi.eqmonitor.domain.OptDeviceMonitorLastData;
import com.ruoyi.eqmonitor.service.IOptDeviceMonitorLastDataService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Profile;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 刷新最后实时数据到数据库
 */
@Slf4j
@Component
@Profile("prod")
@ConditionalOnProperty( value = {"mqtt.enable"}, matchIfMissing = false)
public class FlushLastMonitorToDbTask {

    @Resource
    private RedisService redisService;

    @Resource
    private IOptDeviceMonitorLastDataService deviceMonitorLastDataService;

    /**
     * 3分钟执行一次 将实时数据落库
     * SpringBoot 默认是定时任务同步执行 所以不会出现并发问题
     */
    @Scheduled(cron = "0 0/3 * * * ?")
    public void flushToDb() {
        long start = System.currentTimeMillis();
        log.info("开始更新设备最新实时数据表");
        ScanOptions build = ScanOptions.scanOptions().count(10000).match(MonitorDataMapper.getRedisHKey("*")).build();
        RedisTemplate redisTemplate = redisService.redisTemplate;
        Cursor<String> scan = redisTemplate.scan(build);
        // 聚合所有的hkey
        List<OptDeviceMonitorLastData> cacheList = new ArrayList<>();
        while (scan.hasNext()) {
            String key = scan.next();
            Map<String, OptDeviceMonitorLastData> cacheMap = redisService.getCacheMap(key);
            if (MapUtil.isNotEmpty(cacheMap)) {
                cacheList.addAll(cacheMap.values());
            }
        }
        // 去除所有null元素
        cacheList = cacheList.stream().filter(ObjectUtil::isNotNull).collect(Collectors.toList());
        if (CollectionUtil.isNotEmpty(cacheList)) {
            // 500条数据切为一组
            List<List<OptDeviceMonitorLastData>> split = CollectionUtil.split(cacheList, 500);
            for (List<OptDeviceMonitorLastData> optDeviceMonitorLastData : split) {
                deviceMonitorLastDataService.saveOrUpdateMonitorLastData(optDeviceMonitorLastData);
            }
        }
        log.info("更新设备最新实时数据表完成, 数据条数: {}, 耗时: {}", cacheList.size(), (System.currentTimeMillis() - start));
    }
}
